﻿using System;
using System.Collections;
using System.Collections.Generic;
using Me.Prettyprint.Cassandra.Connection;
using Me.Prettyprint.Cassandra.Model.Thrift;
using Me.Prettyprint.Cassandra.Serializers;
using Me.Prettyprint.Cassandra.Service;
using Me.Prettyprint.Cassandra.Utils;
using Me.Prettyprint.Hector.Api;
using Me.Prettyprint.Hector.Api.Beans;
using Me.Prettyprint.Hector.Api.Query;
using Org.Apache.Cassandra.Thrift;

namespace Me.Prettyprint.Cassandra.Model
{
    /// <summary>
    /// A couple of important notes about CqlQueryOperation and why it's different from
    /// the Java version. Due to differences in Generics between Java and C#, we have
    /// to make CqlQueryOperation<K,N,V> instead of CqlQueryOperation<CqlRows<K,N,V>>.
    /// </summary>
    /// <typeparam name="K"></typeparam>
    /// <typeparam name="N"></typeparam>
    /// <typeparam name="V"></typeparam>
    public class CqlQueryOperation<K,N,V> : Me.Prettyprint.Cassandra.Service.Operation<CqlRows<K,N,V>>
    {
        public static Logger log = new Logger(typeof(CqlQueryOperation<K, N, V>));
        protected bool cql3 = false;

        public CqlQueryOperation(OperationType opType) : base(opType)
        {
        }
        public CqlQueryOperation(OperationType opType, string cqlversion, bool compression,
            byte[] query, ExecutingKeyspace ks, Serializer<K> keySer, Serializer<N> nameSer,
            Serializer<V> valSer, bool suppressKey)
            : base(opType)
        {
            CqlVersion = cqlversion;
            if (cqlversion != null && cqlversion.StartsWith("3"))
            {
                cql3 = true;
            }
            UseCompression = compression;
            Query = query;
            Keyspace = ks;
            KeySerializer = keySer;
            ColumnNameSerializer = nameSer;
            ValueSerializer = valSer;
            SuppressKeyInColumns = suppressKey;
        }
        public string CqlVersion
        {
            get;
            set;
        }
        public bool UseCompression
        {
            get;
            set;
        }
        public byte[] Query
        {
            get;
            set;
        }
        public ExecutingKeyspace Keyspace
        {
            get;
            set;
        }
        public Serializer<K> KeySerializer
        {
            get;
            set;
        }
        public Serializer<N> ColumnNameSerializer
        {
            get;
            set;
        }
        public Serializer<V> ValueSerializer
        {
            get;
            set;
        }
        public bool SuppressKeyInColumns
        {
            get;
            set;
        }
        public HConsistencyLevel ConsistencyLevel
        {
            get;
            set;
        }
        public HConsistencyLevel GetConsistencyLevel()
        {
            if (ConsistencyLevel != null)
            {
                return ConsistencyLevel;
            }
            else
            {
                return HConsistencyLevel.ONE;
            }
        }
        protected static byte[] KEY_BB = StringSerializer.Instance.ToBytes("KEY");
        private List<Column> FilterKeyColumn(CqlRow row)
        {
            if (SuppressKeyInColumns && row.__isset.columns && row.Columns.Count > 0)
            {
                IEnumerator<Column> columnsIterator = row.Columns.GetEnumerator();
                columnsIterator.MoveNext();
                Column column = columnsIterator.Current;
                if (column.Name.Equals(KEY_BB))
                {
                    row.Columns.Remove(column);
                }
            }
            return row.Columns;
        }

        public override CqlRows<K, N, V> Execute(Org.Apache.Cassandra.Thrift.Cassandra.Client cassandra)
        {
            CqlRows<K, N, V> rows = null;
            try
            {
                if (CqlVersion != null)
                {
                    cassandra.set_cql_version(CqlVersion);
                }
                CqlResult result = null;
                if (cql3)
                {
                    result = cassandra.execute_cql3_query(Query, UseCompression ? Compression.GZIP : Compression.NONE, ThriftConverter.consistencyLevel(GetConsistencyLevel()));
                }
                else
                {
                    result = cassandra.execute_cql_query(Query, UseCompression ? Compression.GZIP : Compression.NONE);
                }
                switch (result.Type)
                {
                    case CqlResultType.VOID:
                        rows = new CqlRows<K, N, V>();
                        break;
                    default:
                        if (result.GetRowCount() > 0)
                        {
                            Dictionary<K, List<Column>> ret = new Dictionary<K, List<Column>>();
                            int keyColumn = -1;
                            byte[] keyBytes = new byte[0];
                            for (IEnumerator<CqlRow> rowsIter = result.Rows.GetEnumerator(); rowsIter.MoveNext(); )
                            {
                                CqlRow row = rowsIter.Current;
                                keyBytes = row.Key;
                                // Note that starting with Cassandra 2.x, 2.1 and forward, KEY column is no longer
                                // treated as a special case. It is just another column returned by Cassandra. This
                                // means when CQL3 is used, we have to find that column index and use it to get the
                                // key value.
                                if (cql3)
                                {
                                    List<Column> columns = row.Columns;
                                    if (keyColumn == -1)
                                    {
                                        foreach (Column c in columns)
                                        {
                                            keyColumn++;
                                            string name = StringSerializer.Instance.FromBytes(c.Name);
                                            if (name.ToUpper().Equals("KEY"))
                                            {
                                                keyBytes = c.Value;
                                                break;
                                            }
                                        }
                                    }
                                    else
                                    {
                                        keyBytes = columns[keyColumn].Value;
                                    }
                                }
                                ret[KeySerializer.FromBytes(keyBytes)] = FilterKeyColumn(row);
                            }
                            rows = new CqlRows<K, N, V>((Dictionary<K, List<Column>>)ret, ColumnNameSerializer, ValueSerializer);
                        }
                        else
                        {
                            Dictionary<K, List<Column>> ret = new Dictionary<K, List<Column>>();
                            rows = new CqlRows<K, N, V>((Dictionary<K, List<Column>>)ret, ColumnNameSerializer, ValueSerializer);
                        }
                        break;
                }
            } catch (Exception ex) {
              throw Keyspace.ExceptionsTranslator.translate(ex);
            }
            return rows;
          }

    }
}
